package com.jfplugin.je;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import com.sleepycat.je.Cursor;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.Environment;
import com.sleepycat.je.OperationStatus;
import com.sleepycat.je.Transaction;

/**
 * Bdb处理类
 * 
 * @author farmer
 *
 */
public class BerkeleyDbPro {

	/**
	 * dbConfig
	 */
	private DatabaseConfig dbConfig = null;

	/**
	 * 数据库Map
	 */
	private Map<String, Database> dbMap = new HashMap<String, Database>();

	private Environment env = null;

	private boolean tx = false;

	/**
	 * 构建BdbPro
	 * 
	 * @param bdbPlugin
	 */
	public BerkeleyDbPro(BerkeleyDbPlugin bdbPlugin) {
		this.dbConfig = bdbPlugin.dbConfig;
		this.env = bdbPlugin.env;
		this.tx = bdbPlugin.tx;
	}

	/**
	 * 开始事物
	 * 
	 * @return
	 */
	public synchronized Transaction beginTransaction() {
		return this.tx ? env.beginTransaction(null, null) : null;
	}

	/**
	 * 提交事务
	 * 
	 * @param tx
	 */
	public synchronized void commit(Transaction txn) {
		if (txn != null) {
			txn.commit();
		}
	}

	/**
	 * 获取数据库
	 * 
	 * @param dbName
	 * @return
	 */
	public synchronized Database getDatabase(String dbName) {
		Database db = dbMap.get(dbName);
		if (db == null) {
			db = env.openDatabase(null, dbName, dbConfig);
			dbMap.put(dbName, db);
		}
		return db;
	}
	
	/**
	 * 删除
	 * @param dbName
	 * @param count
	 * @return
	 */
	public synchronized boolean delete(String dbName, int count){
		Database db = getDatabase(dbName);
		DatabaseEntry keyEntry = new DatabaseEntry();
		DatabaseEntry valueEntry = new DatabaseEntry();
		Transaction txn = beginTransaction();
		Cursor cursor = null;
		try {
			cursor = db.openCursor(txn, null);
			OperationStatus status = cursor.getFirst(keyEntry, valueEntry, null);
			int matches = 0;
			while ((++matches <= count) && (status == OperationStatus.SUCCESS)) {
				cursor.delete();
				status = cursor.getNext(keyEntry, valueEntry, null);
			}
		} catch (Exception e) {
			throw new RuntimeException("获取数据失败！", e);
		} finally {
			if(cursor != null){
				cursor.close();
			}
			commit(txn);
		}
		return true;
	}
	
	/**
	 * 获取指定条数
	 * @param dbName
	 * @param count
	 * @return
	 */
	public synchronized List<KV> get(String dbName, int count) {
		List<KV> results = new ArrayList<KV>();
		Database db = getDatabase(dbName);
		DatabaseEntry keyEntry = new DatabaseEntry();
		DatabaseEntry valueEntry = new DatabaseEntry();
		Transaction txn = beginTransaction();
		Cursor cursor = null;
		try {
			cursor = db.openCursor(txn, null);
			OperationStatus status = cursor.getFirst(keyEntry, valueEntry, null);
			int matches = 0;
			while ((++matches <= count) && (status == OperationStatus.SUCCESS)) {
				KV kv = new KV();
				kv.setKey(keyEntry.getData());
				kv.setValue(valueEntry.getData());
				results.add(kv);
				status = cursor.getNext(keyEntry, valueEntry, null);
			}
		} catch (Exception e) {
			throw new RuntimeException("获取数据失败！", e);
		} finally {
			if(cursor != null){
				cursor.close();
			}
			commit(txn);
		}
		return results;
	}

	/**
	 * 删除数据
	 * 
	 * @param dbName
	 * @param key
	 * @return
	 */
	public synchronized boolean delete(String dbName, byte[] key) {
		Database db = getDatabase(dbName);
		Transaction txn = beginTransaction();
		try {
			OperationStatus status = db.delete(txn, new DatabaseEntry(key));
			if (status == OperationStatus.SUCCESS) {
				return true;
			}
		} catch (Exception e) {
			throw new RuntimeException("删除数据失败！", e);
		} finally {
			commit(txn);
		}
		return false;
	}

	/**
	 * 获取数据
	 * 
	 * @param dbName
	 * @param key
	 * @return
	 */
	public synchronized byte[] get(String dbName, byte[] key) {
		Database db = getDatabase(dbName);
		Transaction txn = beginTransaction();
		DatabaseEntry valueEntry = new DatabaseEntry();
		try {
			OperationStatus status = db.get(txn, new DatabaseEntry(key), valueEntry, null);
			if (status == OperationStatus.SUCCESS) {
				return valueEntry.getData();
			}
		} catch (Exception e) {
			throw new RuntimeException("get数据失败！", e);
		} finally {
			commit(txn);
		}
		return null;
	}

	/**
	 * 获取KV
	 * @param dbName
	 * @param key
	 * @return
	 */
	public synchronized KV getKV(String dbName, byte[] key) {
		Database db = getDatabase(dbName);
		Transaction txn = beginTransaction();
		DatabaseEntry valueEntry = new DatabaseEntry();
		KV kv = new KV();
		try {
			OperationStatus status = db.get(txn, new DatabaseEntry(key), valueEntry, null);
			if (status == OperationStatus.SUCCESS) {
				kv.setKey(key);
				kv.setValue(valueEntry.getData());
				return kv;
			}
		} catch (Exception e) {
			throw new RuntimeException("get数据失败！", e);
		} finally {
			commit(txn);
		}
		return null;
	}
	
	/**
	 * 获取数据
	 * 
	 * @param dbName
	 * @param key
	 * @return
	 */
	public synchronized Object get(String dbName, byte[] key , ObjectTupleBinding binding) {
		Database db = getDatabase(dbName);
		Transaction txn = beginTransaction();
		DatabaseEntry valueEntry = new DatabaseEntry();
		try {
			OperationStatus status = db.get(txn, new DatabaseEntry(key), valueEntry, null);
			if (status == OperationStatus.SUCCESS) {
				return binding.entryToObject(valueEntry);
			}
		} catch (Exception e) {
			throw new RuntimeException("get数据失败！", e);
		} finally {
			commit(txn);
		}
		return null;
	}
	
	/**
	 * 
	 * @param dbName
	 * @param kv
	 */
	public synchronized void put(String dbName, KV kv) {
		put(dbName, kv.getKey(), kv.getValue());
		
	}
	
	/**
	 * PUT数据
	 * 
	 * @param dbName
	 * @param key
	 * @param value
	 * @return
	 */
	public synchronized boolean put(String dbName, byte[] key, byte[] value) {
		Database db = getDatabase(dbName);
		Transaction txn = beginTransaction();
		try {
			OperationStatus status = db.put(txn, new DatabaseEntry(key), new DatabaseEntry(value));
			if (status == OperationStatus.SUCCESS) {
				return true;
			}
		} catch (Exception e) {
			throw new RuntimeException("put数据失败！", e);
		} finally {
			commit(txn);
		}
		return false;
	}

	public synchronized boolean put(String dbName , byte[] key , ObjectTupleBinding binding ){
		Database db = getDatabase(dbName);
		Transaction txn = beginTransaction();
		try {
			DatabaseEntry valueEntry = new DatabaseEntry();
			binding.objectToEntry(binding.getObject(), valueEntry);
			OperationStatus status = db.put(txn, new DatabaseEntry(key), valueEntry);
			if (status == OperationStatus.SUCCESS) {
				return true;
			}
		} catch (Exception e) {
			throw new RuntimeException("put数据失败！", e);
		} finally {
			commit(txn);
		}
		return false;
	}
	
	
	/**
	 * 获取指定库的总条数
	 * 
	 * @param dbName
	 * @return
	 */
	public synchronized long count(String dbName) {
		return getDatabase(dbName).count();
	}

	/**
	 * 关闭指定库
	 * 
	 * @param dbName
	 */
	public synchronized void close(String dbName) {
		Database db = getDatabase(dbName);
		db.close();
		dbMap.remove(dbName);
	}
	
	/**
	 * Map操作
	 * @param dbName
	 * @param mapFunction
	 * @param mapDbName
	 * @return
	 */
	public synchronized void map(String dbName , MapFunction mapFunction,String mapDbName){
		Database db = getDatabase(dbName);
		DatabaseEntry keyEntry = new DatabaseEntry();
		DatabaseEntry valueEntry = new DatabaseEntry();
		Transaction txn = beginTransaction();
		Cursor cursor = null;
		long count = count(dbName);
		try {
			cursor = db.openCursor(txn, null);
			OperationStatus status = cursor.getFirst(keyEntry, valueEntry, null);
			long matches = 0;
			while ((++matches <= count) && (status == OperationStatus.SUCCESS)) {
				KV sourcekv = new KV();
				sourcekv.setKey(keyEntry.getData());
				sourcekv.setValue(valueEntry.getData());
				List<KV> mapkvs = mapFunction.map(sourcekv);
				for (KV mapkv : mapkvs) {
					put(mapDbName, mapkv.getKey(), mapkv.getValue());
				}
				status = cursor.getNext(keyEntry, valueEntry, null);
			}
		} catch (Exception e) {
			throw new RuntimeException("获取数据失败！", e);
		} finally {
			if(cursor != null){
				cursor.close();
			}
			commit(txn);
		}
	}
	
	/**
	 * 
	 * @param dbName
	 * @param reduceFunction
	 * @param reduceDbName
	 */
	public synchronized void reduce(String dbName , ReduceFunction reduceFunction,String reduceDbName){
		Database db = getDatabase(dbName);
		DatabaseEntry keyEntry = new DatabaseEntry();
		DatabaseEntry valueEntry = new DatabaseEntry();
		Transaction txn = beginTransaction();
		Cursor cursor = null;
		long count = count(dbName);
		try {
			cursor = db.openCursor(txn, null);
			OperationStatus status = cursor.getFirst(keyEntry, valueEntry, null);
			long matches = 0;
			while ((++matches <= count) && (status == OperationStatus.SUCCESS)) {
				KV mapKv = new KV();
				mapKv.setKey(keyEntry.getData());
				mapKv.setValue(valueEntry.getData());
				KV reduceKv = new KV();
				reduceKv = getKV(reduceDbName, mapKv.getKey());
				reduceKv = reduceFunction.reduce(reduceKv, mapKv);
				put(reduceDbName, reduceKv);
				status = cursor.getNext(keyEntry, valueEntry, null);
			}
		} catch (Exception e) {
			throw new RuntimeException("获取数据失败！", e);
		} finally {
			if(cursor != null){
				cursor.close();
			}
			commit(txn);
		}
	}

	/**
	 * 关闭全部库
	 */
	public synchronized void closeAll() {
		Set<Entry<String, Database>> entrySet = dbMap.entrySet();
		while (entrySet.size() > 0) {
			Iterator<Entry<String, Database>> iterator = entrySet.iterator();
			if (iterator.hasNext()) {
				close(iterator.next().getKey());
			}
		}
	}

}
